package fun.tan90.easy.binlog.plugin.mqtt.config;

import cn.hutool.core.util.RandomUtil;
import com.github.benmanes.caffeine.cache.CacheLoader;
import fun.tan90.easy.binlog.plugin.common.constants.DamiEventConstants;
import fun.tan90.easy.binlog.plugin.common.model.OutputParam;
import fun.tan90.easy.binlog.plugin.mqtt.model.MqttClientParam;
import lombok.extern.slf4j.Slf4j;
import org.checkerframework.checker.nullness.qual.NonNull;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.noear.dami.Dami;

import java.util.Objects;

/**
 * description 动态连接mqtt
 * company 铁人科技
 *
 * @author Li YuLong
 * @date 2023/8/17 10:56
 */
@Slf4j
public class MqttClientCacheLoader implements CacheLoader<String, MqttClient> {

    @Override
    public @Nullable MqttClient load(@NonNull String outputCode) throws Exception {
        OutputParam outputParam = Dami.<String, OutputParam>bus().sendAndRequest(DamiEventConstants.OUTPUT_PARAM_TOPIC, outputCode);
        if (Objects.isNull(outputParam)) {
            return null;
        }
        MqttClientParam mqttClientParams = (MqttClientParam) outputParam;
        MqttClient client = new MqttClient(mqttClientParams.getBroker(), mqttClientParams.getClientIdPrefix() + RandomUtil.randomNumbers(6), new MemoryPersistence());

        MqttConnectOptions connOpts = new MqttConnectOptions();
        connOpts.setUserName(mqttClientParams.getUsername());
        connOpts.setPassword(mqttClientParams.getPassword().toCharArray());
        // 保留会话
        connOpts.setCleanSession(true);
        // 配置自动重连
        connOpts.setAutomaticReconnect(true);
        // MQTT协议版本
        connOpts.setMqttVersion(mqttClientParams.getMqttVersion());
        // keep alive
        connOpts.setKeepAliveInterval(mqttClientParams.getKeepAliveInterval());
        // 设置回调
        client.setCallback(new MqttCallbackExtended() {
            @Override
            public void connectionLost(Throwable cause) {
                log.error("连接断开，可以做重连【{}】", cause.getMessage());
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                log.info("deliveryComplete messageId {}", token.getMessageId());
            }

            @Override
            public void connectComplete(boolean reconnect, String serverUri) {
                log.info("connectComplete reconnect {} serverUri {}", reconnect, serverUri);
            }
        });
        client.connect(connOpts);
        return client;
    }
}
